package zkclient

import (
	"EasyData/global"
	log "code.google.com/p/log4go"
	"fmt"
	"errors"
	"EasyData/zkclient/basic"
)

/**
	启动zk任务管理
 */
func InitZkTask(){
	//创建根节点
	createNodeIfExists(rootNode,"data-from-mysql-to-hbase-hbas")
	//创建服务的命名空间节点
	createNodeIfExists(getNameSpaceNode(), global.Config().Namespace)
	//创建并监听任务节点的具体配置项
	createAndWatcherTaskNode()
}

/**
	创建作业（永久节点）
 */
func createNodeIfExists(node string,value string) {
	isExist,_,err := basic.GetZkClient().Exists(node)
	if err!= nil {
		errStr := fmt.Sprintf("zookeeper Exists Node [%s] err : %s",node,err)
		log.Error(errStr)
		global.PanicError(errors.New(errStr),nil)
	}
	if !isExist {
		err := basic.CreateForeverZkNode(node,value)
		if err != nil {
			errStr := fmt.Sprintf("zookeeper create Node [%s] err : %s",node,err)
			log.Error(errStr)
			global.PanicError(errors.New(errStr),nil)
		}
	}
}
/**
	如果节点存在，先删除，再创建
	不存在，则创建
 */
func createOrSet(node string, value string) {
	isExist,_,err := basic.GetZkClient().Exists(node)
	if err!= nil {
		errStr := fmt.Sprintf("zookeeper Exists Node [%s] err : %s",node,err)
		log.Error(errStr)
		return
	}
	if !isExist {
		err := basic.CreateForeverZkNode(node,value)
		if err != nil {
			errStr := fmt.Sprintf("zookeeper create Node [%s] err : %s",node,err)
			log.Error(errStr)
		}
	}else{
		err := basic.DeleteZkNode(node)
		if err != nil {
			errStr := fmt.Sprintf("zookeeper delete Node [%s] err : %s",node,err)
			log.Error(errStr)
		}
		err = basic.CreateForeverZkNode(node,value)
		if err != nil {
			errStr := fmt.Sprintf("zookeeper create Node [%s] err : %s",node,err)
			log.Error(errStr)
		}
	}
}
/**
	创建Session类型的节点
 */
func createSessionNodeIfExists(node string,value string) {

	isExist,_,err := basic.GetZkClient().Exists(node)
	if err!= nil {
		errStr := fmt.Sprintf("zookeeper Exists Node [%s] err : %s",node,err)
		log.Error(errStr)
		return
	}
	if !isExist {
		err := basic.CreateSessionZkNode(node,value)
		if err != nil {
			errStr := fmt.Sprintf("zookeeper create Node [%s] err : %s",node,err)
			log.Error(errStr)
		}
	}
}
/**
	创建	session节点，有则删除再创建
 */
func createSetSessionNode(node string,value string){
	isExist,_,err := basic.GetZkClient().Exists(node)
	if err!= nil {
		errStr := fmt.Sprintf("zookeeper Exists Node [%s] err : %s",node,err)
		log.Error(errStr)
		return
	}
	if !isExist {
		err := basic.CreateSessionZkNode(node,value)
		if err != nil {
			errStr := fmt.Sprintf("zookeeper create Node [%s] err : %s",node,err)
			log.Error(errStr)
		}
	}else{
		err := basic.DeleteZkNode(node)
		if err != nil {
			errStr := fmt.Sprintf("zookeeper delete Node [%s] err : %s",node,err)
			log.Error(errStr)
		}
		err = basic.CreateSessionZkNode(node,value)
		if err != nil {
			errStr := fmt.Sprintf("zookeeper create Node [%s] err : %s",node,err)
			log.Error(errStr)
		}
	}
}

func createAndWatcherTaskNode() {

	//和任务节点同名节点下，创建内存值节点和协程数节点
	go createIfNoExistsMemAndGetGoroutine()
	//和任务节点同名节点下，创建或更新mysql的ip配置信息和Hbase的ip等信息
	createOrSet(getTaskNode(mysql),fmt.Sprintf("%s:%s",global.Config().Mysql.Ip,global.Config().Mysql.Port))
	createOrSet(getTaskNode(hbase),fmt.Sprintf("%s",global.Config().Hbase.Ip))

	tasks :=global.Config().Tasks
	for index := range tasks {
		var task *global.Task
		task = tasks[index]
		if task.IsUse {
			//创建任务节点
			createNodeIfExists(getTaskNode(task.TaskName),task.TaskName)
			//开始配置每个节点的具体项配置
			createChildSessionNodeIfExists(task)
		}
	}
}

/**
	获取每个作业名称NameSpace节点
 */
func getNameSpaceNode() (node string){
	return fmt.Sprintf("%s/%s",rootNode,global.Config().Namespace)
}
/**
	获取每个作业名称的Task节点
 */
func getTaskNode(taskName string)  (node string){
	return fmt.Sprintf("%s/%s",getNameSpaceNode(),taskName)
}
/**
	获取每个作业名称的Task节点的具体配置项节点名
 */
func getTaskConfigNode(taskName string,config string)(node string) {
	return fmt.Sprintf("%s/%s",getTaskNode(taskName),config)
}